package com.diego.strom.demo.wordcount;

import java.util.HashMap;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

/**
 * 对单词计数
 * 
 * @author Manfred
 *
 */
public class WordCounterRichBolt implements IRichBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private Integer id;  
	private String name;  
	private Map<String, Integer> counters;  
    private OutputCollector collector;  

	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.counters = new HashMap<String, Integer>();  
        this.collector = collector;  
        this.name = context.getThisComponentId();  
        this.id = context.getThisTaskId();  
	}

	public void execute(Tuple input) {
		String str = input.getString(0);  
        if (!counters.containsKey(str)) {  
            counters.put(str, 1);  
        } else {  
            Integer c = counters.get(str) + 1;  
            counters.put(str, c);  
        }  
        // 确认成功处理一个tuple  
        collector.ack(input);  
	}
	 /** 
     * Topology执行完毕的清理工作，比如关闭连接、释放资源等操作都会写在这里 
     * 因为这只是个Demo，我们用它来打印我们的计数器 
     * */  
	public void cleanup() {
		System.out.println("-- Word Counter [" + name + "-" + id + "] --");  
        for (Map.Entry<String, Integer> entry : counters.entrySet()) {  
            System.out.println(entry.getKey() + ": " + entry.getValue());  
        }  
        counters.clear();  
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		 declarer.declare(new Fields("word"));  
	}

	public Map<String, Object> getComponentConfiguration() {

		return null;
	}

}
